// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: fengyang

#include <algorithm>
#include <map>
#include <set>
#include <vector>

#include "coin/proto/symbology.pb.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/base/string_util.h"
#include "coin2/exchange/base/executor/topic_publisher.h"
#include "coin2/exchange/lmax_digital/feed_v1/fix_connection.h"

#include <boost/date_time/posix_time/time_formatters.hpp>

#include "quickfix/Session.h"
#include "quickfix/SessionID.h"


namespace coin2::exchange::lmax_digital::feed_v1 {

FixConnection::FixConnection(
  base::executor::LiveAsyncExecutor* exec,
  const WsConnectionConfig& config)
    : exec_(exec),
      scheduler_(new Scheduler(exec_->RawIoc())),
      config_(config) {
  // bind local ip before initiator
  // SOCKET_CONNECT_SOURCE_HOST
  // SOCKET_CONNECT_SOURCE_PORT
  // port setting to session setttings

  constexpr std::string_view fix_protocol("fix://");
  CHECK(config_.url.host.rfind(fix_protocol) == 0)
      << "Not FIX config " << config_.url.host;

  std::string config_path = config_.url.host.substr(fix_protocol.size());
  FIX::SessionSettings original_settings(config_path);
  session_settings_ = std::make_shared<FIX::SessionSettings>();
  session_settings_->set(original_settings.get());
  std::set<FIX::SessionID> session_ids = original_settings.getSessions();

  for (auto& session_id : session_ids) {
    FIX::Dictionary dict = original_settings.get(session_id);
    if (!config_.url.bind_ip.empty()) {
      dict.setString("SocketConnectSourceHost", config_.url.bind_ip);
    }
    session_settings_->set(session_id, dict);
  }
}

FixConnection::~FixConnection() {
  scheduler_->Stop();
}

std::shared_ptr<FIX::SessionSettings> FixConnection::GetSessionSettings() {
  return session_settings_;
}

void FixConnection::onCreate(const FIX::SessionID& sessionID) {
  LOG(INFO) << "Fix create";

  // Schedule sending periodic ping if needed.
  std::weak_ptr<FixConnection> self(shared_from_this());
  if (config_.mea_config->periodic_force_reconnect_minute > 0) {
    scheduler_->Start();
    scheduler_->AddPeriodicTask(
        [self] {
          if (auto s = self.lock()) {
            s->Resubscribe();
          } else {
            LOG(ERROR) << "boost::asio handler object (FixConnection) does not exist";
          }
        },
        config_.mea_config->periodic_force_reconnect_minute * 60 * 1'000'000'000LL,
        "force subscribe");
  }
}

void FixConnection::onLogon(const FIX::SessionID& sessionID) {
  LOG(INFO) << "Fix logon";
  default_session_ = sessionID;
  is_logon_ = true;
  // subscribe message send
  FIX42::MarketDataRequest message;
  SendSubscribeRequest(&message, sessionID);
}

void FixConnection::onLogout(const FIX::SessionID& sessionID) {
  LOG(INFO) << "Fix logout " << sessionID.toString();
  is_logon_ = false;
}

void FixConnection::toAdmin(FIX::Message& message, const FIX::SessionID& sessionID) {
  crack(message, sessionID);
}

void FixConnection::toApp(FIX::Message& message, const FIX::SessionID& sessionID) {
  LOG(INFO) << "Fix to app";
  std::string content = message.toString();
  ToPrintMessage(&content);
  LOG(INFO) << content;
}

void FixConnection::fromAdmin(const FIX::Message& message, const FIX::SessionID& sessionID) {
  crack(message, sessionID);
  std::string content = message.toString();
  ToPrintMessage(&content);
  LOG(INFO) << content;
}

void FixConnection::fromApp(const FIX::Message& message, const FIX::SessionID& sessionID) {
  std::string fix_data = message.toString();
  auto packet = std::shared_ptr<base::executor::TopicRecordPacket>(
    new base::executor::TopicRecordPacket{
        config_.topic,
        0,
        base::executor::TopicRecordProcessor::GetTimestamp(),
        0,
        0,
        0,
        0,
        0,
        0,
        fix_data});
  exec_->injector().push(packet);
}

void FixConnection::onMessage(FIX42::Logon& message, const FIX::SessionID& sessionID) {
  FIX::MsgSeqNum seq_num;
  std::string seq_num_val = message.getHeader().getField(seq_num.getTag());
  msg_seq_num_ = std::stoi(seq_num_val);
  BuildLogonRequest(&message, sessionID);
}

void FixConnection::onMessage(FIX42::TestRequest& message, const FIX::SessionID& sessionID) {
  FIX42::Heartbeat heartbeat;
  FIX::TestReqID req_id;
  std::string test_req_id = message.getField(req_id.getTag());
  heartbeat.setField(FIX::TestReqID(test_req_id));
  BuildHeaderTrailer(&heartbeat, sessionID, "0");
  FIX::Session::sendToTarget(heartbeat);
}

void FixConnection::onMessage(FIX42::ResendRequest& message, const FIX::SessionID& sessionID) {
  std::string content = message.toString();
  ToPrintMessage(&content);
  LOG(INFO) << content;
}

void FixConnection::onMessage(FIX42::SequenceReset& message, const FIX::SessionID& sessionID) {
  std::string content = message.toString();
  ToPrintMessage(&content);
  LOG(INFO) << content;
}

void FixConnection::onMessage(FIX42::Reject& message, const FIX::SessionID& sessionID) {
  std::string content = message.toString();
  ToPrintMessage(&content);
  LOG(INFO) << content;
}

void FixConnection::BuildLogonRequest(
    FIX::Message* message,
    const FIX::SessionID& sessionID) {
  // logon message send
  FIX::SenderCompID comp_id;
  std::string comp_id_val = message->getHeader().getField(comp_id.getTag());
  std::string raw_data = session_settings_->get(sessionID).getString("Password");
  message->setField(FIX::RawDataLength(raw_data.size()));
  message->setField(FIX::RawData(raw_data));
  FIX::Header& header =  message->getHeader();
  // FIX::MsgSeqNum stamp automatically
  header.setField(FIX::PossDupFlag(FIX::PossDupFlag_YES));
  header.setField(FIX::OrigSendingTime());
  header.setField(FIX::BodyLength(message->bodyLength()));
  FIX::Trailer& trailer = message->getTrailer();
  trailer.setField(FIX::CheckSum(message->checkSum()));
}

void FixConnection::SendSubscribeRequest(
    FIX42::MarketDataRequest* message,
    const FIX::SessionID& sessionID)  {
  for (auto& symbol : config_.symbols) {
    BuildSubscribeRequest(message, sessionID, symbol);
    FIX::Session::sendToTarget(*message);
  }
}

void FixConnection::BuildSubscribeRequest(
    FIX42::MarketDataRequest* message,
    const FIX::SessionID& sessionID,
    const std::string& symbol) {
  static const base::symbology::ProductInfoConstant kLmaxdigitalProductInfo(
      "data/coin2/product_info/Spot.Lmaxdigital.v1.json");

  message->setField(FIX::MDReqID(Iso8601FromTimestamp(GetCurrentTimestamp())));
  message->setField(FIX::SubscriptionRequestType('1'));  // SNAPSHOT PLUS UPDATES
  message->setField(FIX::MDUpdateType(1));  // incremental refresh
  message->setField(FIX::MarketDepth(
      session_settings_->get(sessionID).getInt("MarketDepth")));

  FIX42::MarketDataRequest::NoMDEntryTypes entry_group;
  for (auto& channel : config_.url.channels) {
    const auto& tmpl = config_.mea_config->channels.at(channel);
    std::vector<std::string> channels = Split(tmpl, ",");
    for (auto& book_type : channels) {
      if ("BID" == book_type) {
        entry_group.setField(FIX::MDEntryType('0'));
        message->addGroup(entry_group);
      } else if ("OFFER" == book_type) {
        entry_group.setField(FIX::MDEntryType('1'));
        message->addGroup(entry_group);
      } else if ("TRADE" == book_type) {
        // not publish trade data in UAT
        entry_group.setField(FIX::MDEntryType('2'));
        message->addGroup(entry_group);
      } else {
        CHECK(false) << "Not support MDEntryType " << book_type;
      }
    }
  }

  auto pi_proto = kLmaxdigitalProductInfo.GetProductInfoByNativeSymbol(symbol);
  std::string id = pi_proto.id();

  FIX42::MarketDataRequest::NoRelatedSym symbol_group;
  symbol_group.setField(FIX::Symbol(symbol));
  symbol_group.setField(FIX::SecurityID(id));
  symbol_group.setField(FIX::IDSource("8"));  // exchange symbol
  message->addGroup(symbol_group);

  BuildHeaderTrailer(message, sessionID, "V");
}

void FixConnection::BuildHeaderTrailer(
    FIX::Message* message,
    const FIX::SessionID& session_id,
    const std::string& msg_type) {
  // set header
  FIX::Header& header =  message->getHeader();
  header.setField(FIX::BeginString("FIX.4.2"));
  header.setField(FIX::BodyLength(message->bodyLength()));
  header.setField(FIX::MsgType(msg_type));
  std::string sender_id = session_settings_->get(session_id).getString("SenderCompID");
  std::string target_id = session_settings_->get(session_id).getString("TargetCompID");
  header.setField(FIX::SenderCompID(sender_id));
  header.setField(FIX::TargetCompID(target_id));
  header.setField(FIX::MsgSeqNum(++msg_seq_num_));
  header.setField(FIX::PossDupFlag(FIX::PossDupFlag_YES));
  header.setField(FIX::SendingTime());
  header.setField(FIX::OrigSendingTime());
  FIX::Trailer& trailer = message->getTrailer();
  trailer.setField(FIX::CheckSum(message->checkSum()));
}

void FixConnection::ToPrintMessage(std::string* message) {
  std::replace(message->begin(), message->end(), '\001', '|');
}

void FixConnection::Resubscribe() {
  if (is_logon_) {
    FIX42::MarketDataRequest message;
    SendSubscribeRequest(&message, default_session_);
  }
}

}  // namespace coin2::exchange::lmax_digital::feed_v1
